package com.allen.study.test_common.utils.datax;

import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.date.StopWatch;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.allen.study.test_common.exception.CustomRuntimeException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.IOException;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @ClassName: DataXUtils
 * @Author: AllenSun
 * @Date: 2025/3/4 23:49
 */
@Slf4j
@Component
@RefreshScope
public class DataXUtils {

    @Value("${datax.home}")
    private String dataxHomePath ;

    @Value("${datax.jsonFilePath}")
    private String jsonFilePath ;

    // TODO 支持动态传参mysql模板执行
    // TODO 多写几个模板案例
    // TODO 加入线程池

    public void syncDataxMysql2MysqlByCommand (DataSourceConfig source,
                                      DataSourceConfig target) {
        String jobCommand = syncMysqlToMysqlJobCommand(source, target);
        try {
            // 执行数据同步任务
            String result = this.executeDataXCommand(jobCommand);
            log.info("数据同步结果：" + result);
        } catch (Exception e) {
            e.printStackTrace();
            log.info("数据同步失败：" + e.getMessage());
            throw new CustomRuntimeException("数据同步失败：" + e.getMessage());
        }
    }

    public void syncDataxMysql2MysqlByJsonFile (DataSourceConfig source,
                                      DataSourceConfig target,String jobFileName) {
        // String jobFileName = "testCreateMysqlJob.json";
        StopWatch stopWatch = new StopWatch("datax同步数据耗时");
        stopWatch.start("开始");
        try {
            // 生成 Job 文件
            JSONObject job = generateMysqlToMysqlJobJson(source, target);
            this.generateJobFile(job, jobFileName);
            // 执行数据同步任务
            this.executeDataXJob(jobFileName);
        } catch (IOException e) {
            e.printStackTrace();
            log.info("数据同步失败：" + e.getMessage());
            throw new CustomRuntimeException("数据同步失败：" + e.getMessage());
        }
        stopWatch.stop();
        log.info(stopWatch.prettyPrint(TimeUnit.SECONDS));

    }



    /**
     * 生成 MySQL 到 MySQL 的 DataX Job JSON 配置
     * @return Job 的 JSON 配置
     */
    public JSONObject generateMysqlToMysqlJobJson(DataSourceConfig source,
                                                         DataSourceConfig target) {

        String sourceUrl=source.getJdbcUrl();
        String sourceUsername=source.getUser();
        String sourcePassword=source.getPassword();
        String sourceTable=source.getTable();
        String targetUrl=target.getJdbcUrl();
        String targetUsername=target.getUser();
        String targetPassword=target.getPassword();
        String targetTable=target.getTable();

        List<String> preSql = ListUtil.of("TRUNCATE TABLE "+targetTable);

        // 查询表字段
        List<String> sourceColumns = getTableColumns(sourceUrl, sourceUsername, sourcePassword, sourceTable);
        List<String> targetColumns = getTableColumns(targetUrl, targetUsername, targetPassword, targetTable);
        List<String> combineColumns = sourceColumns.stream().filter(it->targetColumns.contains(it)).collect(Collectors.toList());

        if(ObjectUtil.isEmpty(combineColumns)){
            throw new CustomRuntimeException("源表和目标表不存在匹配字段");
        }

        JSONObject job = new JSONObject();
        JSONObject setting = new JSONObject();
        JSONObject speed = new JSONObject();
        speed.put("channel", "3");
        setting.put("speed", speed);
        job.put("setting", setting);

        JSONObject content = new JSONObject();
        JSONObject reader = new JSONObject();
        reader.put("name", "mysqlreader");
        JSONObject readerParameter = new JSONObject();
        readerParameter.put("username", sourceUsername);
        readerParameter.put("password", sourcePassword);
        JSONArray connection = new JSONArray();
        JSONObject conn = new JSONObject();
        JSONArray jdbcUrl = new JSONArray();
        jdbcUrl.add(sourceUrl);
        conn.put("jdbcUrl", jdbcUrl);
        conn.put("table", new String[]{sourceTable});
        connection.add(conn);
        readerParameter.put("connection", connection);
        readerParameter.put("column", combineColumns);
        reader.put("parameter", readerParameter);

        JSONObject writer = new JSONObject();
        writer.put("name", "mysqlwriter");
        JSONObject writerParameter = new JSONObject();
        writerParameter.put("username", targetUsername);
        writerParameter.put("password", targetPassword);
        writerParameter.put("connection", new JSONArray() {{
            add(new JSONObject() {{
                put("jdbcUrl", targetUrl);
                put("table", new String[]{targetTable});
            }});
        }});
        writerParameter.put("column", combineColumns);
        writerParameter.put("preSql", preSql);// 前置执行sql，先清空表再执行同步数据逻辑
        writer.put("parameter", writerParameter);

        content.put("reader", reader);
        content.put("writer", writer);

        JSONArray contentArray = new JSONArray();
        contentArray.add(content);
        job.put("content", contentArray);

        JSONObject allJobContent = new JSONObject();
        allJobContent.put("job",job);

        return allJobContent;
    }

    /**
     * 生成 MySQL 到 MySQL 的 DataX Job JSON 配置
     * @return Job 的 JSON 配置
     */
    public String syncMysqlToMysqlJobCommand(DataSourceConfig source,
                                                   DataSourceConfig target) {

        String sourceUrl=source.getJdbcUrl();
        String sourceUsername=source.getUser();
        String sourcePassword=source.getPassword();
        String sourceTable=source.getTable();
        String targetUrl=target.getJdbcUrl();
        String targetUsername=target.getUser();
        String targetPassword=target.getPassword();
        String targetTable=target.getTable();

        // 查询表字段
        List<String> sourceColumns = getTableColumns(sourceUrl, sourceUsername, sourcePassword, sourceTable);
        List<String> targetColumns = getTableColumns(targetUrl, targetUsername, targetPassword, targetTable);
        List<String> combineColumns = sourceColumns.stream().filter(it->targetColumns.contains(it)).collect(Collectors.toList());

        if(ObjectUtil.isEmpty(combineColumns)){
            log.error("源表和目标表不存在匹配字段");
            throw new CustomRuntimeException("源表和目标表不存在匹配字段");
        }

        StringBuilder columnStr = new StringBuilder();
        for (int i = 0; i < combineColumns.size(); i++) {
            if (i > 0) {
                columnStr.append(",");
            }
            columnStr.append("\"").append(combineColumns.get(i)).append("\"");
        }

        String readerJson = "{\"name\":\"mysqlreader\",\"parameter\":{\"username\":\"" + sourceUsername + "\",\"password\":\"" + sourcePassword + "\",\"connection\":[{\"jdbcUrl\":[\"" + sourceUrl + "\"],\"table\":[\"" + sourceTable + "\"]}],\"column\":[" + columnStr + "]}}";
        String writerJson = "{\"name\":\"mysqlwriter\",\"parameter\":{\"username\":\"" + targetUsername + "\",\"password\":\"" + targetPassword + "\",\"connection\":[{\"jdbcUrl\":[\"" + targetUrl + "\"],\"table\":[\"" + targetTable + "\"]}],\"column\":[" + columnStr + "]}}";
        String contentJson = "{\"reader\":" + readerJson + ",\"writer\":" + writerJson + "}";
        String jobJson = "{\"job\":{\"setting\":{\"speed\":{\"channel\":\"3\"}},\"content\":[" + contentJson + "]}}";

        // TODO mac电脑自带python3环境
        // String finalCommand =  "python3 " + dataxHomePath + "/bin/datax.py '" + jsonFilePath + jobJson + "'";
        String finalCommand =
                "python3 " + dataxHomePath + "/bin/datax.py  -r streamreader -w streamwriter '" + jobJson + "'";
        log.info("DataX完整命令：{}",finalCommand);
        return finalCommand;
    }

    /**
     * 查询 MySQL 表的字段
     * @param url 数据库连接 URL
     * @param username 数据库用户名
     * @param password 数据库密码
     * @param tableName 表名
     * @return 字段列表
     * @throws SQLException SQL 异常
     */
    public List<String> getTableColumns(String url, String username, String password, String tableName) {
        List<String> columns = new ArrayList<>();
        try (Connection connection = DriverManager.getConnection(url, username, password);
             Statement statement = connection.createStatement();
             ResultSet resultSet = statement.executeQuery("SHOW COLUMNS FROM " + tableName)) {
            while (resultSet.next()) {
                columns.add(resultSet.getString("Field"));
            }
        } catch (Exception e) {
            log.error("查询表字段失败：{}",e.getMessage());
            throw new CustomRuntimeException("查询表字段失败");
        }
        return columns;
    }

    /**
     * 生成 DataX Job 文件
     * @param jobJson Job 的 JSON 配置
     * @param filePath Job 文件保存路径
     * @throws IOException 文件操作异常
     */
    public void generateJobFile(JSONObject jobJson, String filePath) throws IOException {
        String fullFilePath = jsonFilePath+filePath;
        File jobFile = new File(fullFilePath);
        FileUtils.writeStringToFile(jobFile, jobJson.toJSONString(), "UTF-8");
    }

    /**
     * 执行 DataX 数据同步任务
     * @param jobFileName Job 文件的名称
     * @return 执行结果
     * @throws IOException 执行命令异常
     * @throws InterruptedException 线程中断异常
     */
    public String executeDataXJob(String jobFileName) {
        String command =
                "python3 " + dataxHomePath + "/bin/datax.py " + jsonFilePath + jobFileName;
        log.info("执行Datax命令：{}",command);
        StringBuilder result = new StringBuilder();
        try {
            Process process = Runtime.getRuntime().exec(command);
            java.io.InputStream is = process.getInputStream();
            java.io.BufferedReader reader = new java.io.BufferedReader(new java.io.InputStreamReader(is));
            String line;
            while ((line = reader.readLine()) != null) {
                result.append(line).append("\n");
            }
            process.waitFor();
        } catch (Exception e) {
            log.info("datax命令执行失败：{}",e.getMessage());
            throw new CustomRuntimeException("datax命令执行失败："+e.getMessage());
        }
        log.info("datax命令执行结果：{}",result);
        return result.toString();
    }

    /**
     * 执行 DataX 数据同步任务
     * @param command DataX 执行命令
     * @return 执行结果
     * @throws Exception 执行命令异常
     */
    public String executeDataXCommand(String command) throws Exception {
        Process process = Runtime.getRuntime().exec(command);
        java.io.InputStream is = process.getInputStream();
        java.io.BufferedReader reader = new java.io.BufferedReader(new java.io.InputStreamReader(is));
        StringBuilder result = new StringBuilder();
        String line;
        while ((line = reader.readLine()) != null) {
            result.append(line).append("\n");
        }
        process.waitFor(10, TimeUnit.MINUTES);
        return result.toString();
    }


}
